package com.darchrow.transaction;

/**
 * @author mdl
 * @date 2020/4/27 14:52
 */

import com.alibaba.fastjson.JSON;
import com.darchrow.Constants;
import com.darchrow.mapper.TxLogMapper;
import com.darchrow.pojo.Order;
import com.darchrow.pojo.TxLog;
import com.darchrow.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

@Slf4j
@Component
@RocketMQTransactionListener(txProducerGroup = Constants.MQ_PRODUCER_GROUP)
public class ProducerTransactionListener implements RocketMQLocalTransactionListener {

    @Resource
    private OrderService orderService;
    @Resource
    private TxLogMapper txLogMapper;

    /**
     * 事务消息发送mq成功后的回调方法
     *
     * @param msg
     * @param arg
     * @return 返回事务状态
     * RocketMQLocalTransactionState.COMMIT：提交事务，提交后broker才允许消费者使用
     * RocketMQLocalTransactionState.ROLLBACK：回滚事务，回滚后消息将被删除，并且不允许别消费
     * RocketMQLocalTransactionState.Unknown：中间状态，表示MQ需要核对，以确定状态
     */
    @Transactional(rollbackFor = Exception.class)
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        try {
            String str = new String((byte[]) msg.getPayload());
            Order order = JSON.parseObject(str, Order.class);

            orderService.createOrder(order);
            // 当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit，mq将消息的状态改为可消费
            // 提交UNKNOWN，可触发check事务状态回查
//            return RocketMQLocalTransactionState.UNKNOWN;
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * 事务状态回查，查询是否下单成功
     * 注意：
     * RocketMQ在 V3.2.6之前的版本这样写就可以了，但是之后的版本被关于事务回查这个借口被阉割了，不会在进行事务回查操作。
     * 回查要建回溯表
     *
     * @param msg
     * @return 返回事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("事务状态回查={}", msg);
        String str = new String((byte[]) msg.getPayload());
        Order order = JSON.parseObject(str, Order.class);
        //事务id
        String txNo = order.getTxNum();
        TxLog txLog = txLogMapper.selectById(txNo);
        if (txLog != null) {
            log.info("发现此消息的事务日志已提交，则提交消息，让消费者可见={}", msg);
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
